package com.tengju.data.infrastructure.kafka.consumer;

import com.tengju.data.domain.model.message.FlatMessage;
import com.tengju.data.domain.shared.JsonUtil;
import com.tengju.data.infrastructure.shared.InfraException;
import com.tengju.user.application.user.UserDocumentJobService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
@Slf4j
@ConditionalOnProperty(prefix = "tengju.binlog.es.consumer", name = "switch", havingValue = "true")
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class TengjuBinlogESConsumer {

    private final UserDocumentJobService userDocumentJobService;

    @KafkaListener(
            topics = "#{'${kafka.listener.topics.tengju.binlog.es}'.split(',')}",
            containerFactory = "tjBinlogKafkaListenerContainerFactory",
            errorHandler = "binlogConsumerListenerErrorHandler"
    )
    public void listener(ConsumerRecord<String, String> record, Acknowledgment ack)
            throws InfraException {
        // TODO 方法抽象
        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isEmpty()) {
            throw new InfraException("received kafkaMessage empty");
        }
        String binlog = kafkaMessage.get();
        log.info("receive tengju binlog: {}", binlog);
        FlatMessage flatMessage = JsonUtil.toObject(binlog, FlatMessage.class);
        if (flatMessage == null) {
            throw new InfraException("binlog toObject error");
        }
        String type = flatMessage.getType();
        if (StringUtils.isBlank(type)) {
            throw new InfraException("binlog type is empty");
        }
        switch (type) {
            case "INSERT":
            case "UPDATE":
            case "DELETE":
                save(flatMessage);
                break;
            default:
                log.warn("type not support: {}", type);
                break;
        }
        ack.acknowledge();
    }

    private void save(FlatMessage flatMessage) {
        String primaryKeyName = flatMessage.getPkNames().get(0);
        String tableName = flatMessage.getTable();
        String primaryKey = flatMessage.getData().get(primaryKeyName);
        try {
            switch (tableName) {
                case "user_info":
                    userDocumentJobService.saveUserInfo(primaryKey);
                    break;
                case "motivation_star_sign_info":
                    userDocumentJobService.saveMotivationStarSignInfo(primaryKey);
                    break;
                case "motivation_star_change_record":
                    userDocumentJobService.saveMotivationStarChangeRecord(primaryKey);
                    break;
                case "user_domination_member_note":
                    userDocumentJobService.saveDominationMemberNote(primaryKey);
                    break;
                case "w_user_brand_provider":
                    userDocumentJobService.saveUserBrandProvider(primaryKey);
                    break;
                case "w_user":
                    userDocumentJobService.saveUser(primaryKey);
                    break;
                default:
                    log.warn("tableName not support");
            }
        } catch (Exception e) {
            log.error("sync es error table: {}, primaryKey: {}, error", tableName, primaryKey, e);
            throw new InfraException("tengju binlog sync es error", e);
        }
    }
}
